/*
 * TcpsRtpSocket.cpp
 *
 *  Created on: 2018年5月1日
 *      Author: zhengboyuan
 */

#include "TcpsRtpSocket.h"
#include "RtpHeader.h"
#include "RtpPort.h"
#include "CLog.h"


namespace av
{

TcpsRtpSocket::TcpsRtpSocket():
		m_offset(),
		m_state(kDisconnected)
{
}

TcpsRtpSocket::~TcpsRtpSocket()
{
	close();
}

bool TcpsRtpSocket::open(const char* ip, int port)
{
	CLog::info("TcpsRtpSocket::open. %s:%d\n", ip, port);

	if (!m_listenSocket.open(SOCK_STREAM))
	{
		return false;
	}

	m_listenSocket.setReuse();
	m_listenSocket.setNonblock(true);

	bool done = false;
	if (port == 0)
	{
		for (size_t i = 0; i < (RtpPort::count() + 1); ++ i)
		{
			port = RtpPort::make();

			comn::SockAddr addr(ip, port);
			if (0 == m_listenSocket.bind(addr))
			{
				m_addr = addr;
				done = true;
				break;
			}
		}
	}
	else
	{
		comn::SockAddr addr(ip, port);
		if (0 == m_listenSocket.bind(addr))
		{
			m_addr = addr;
			done = true;
		}
	}

	if (!done)
	{
		close();
		return false;
	}

	m_listenSocket.listen();

	m_socketPair.open();

	start();

	CLog::info("TcpsRtpSocket::open done. port: %d\n", m_addr.getPort());
	return true;
}

void TcpsRtpSocket::close()
{
	if (isRunning())
	{
		stop();
	}

	m_listenSocket.close();

	m_connSocket.close();

	m_socketPair.close();
}

bool TcpsRtpSocket::isOpen()
{
	return m_listenSocket.isOpen();
}

NetAddress TcpsRtpSocket::getLocalAddr()
{
	return NetAddress(m_addr.getIPAddr(), m_addr.getPort());
}

bool TcpsRtpSocket::setPeer(const char* ip, int port)
{
	 // pass
	return true;
}

bool TcpsRtpSocket::send(RtpPacket& packet, int ms)
{
	if (!checkConnect())
	{
		return false;
	}

	if (!flushWrite(ms))
	{
		return false;
	}

	bool done = false;

	m_offset = 0;

	m_buffer.ensure(packet.size + sizeof(RTP_FIXED_HEADER));

	RTP_FIXED_HEADER* header = (RTP_FIXED_HEADER*)m_buffer.data();
	header->version = 2;
	header->marker = packet.mark;

	header->timestamp = htonl(m_initTime + packet.ts);

	header->ssrc = htonl(m_ssrc);
	header->seq_no = htons(m_seq ++);
	header->payload = packet.pt;

	uint8_t* data = m_buffer.data() + sizeof(RTP_FIXED_HEADER);
	memcpy(data, packet.data, packet.size);

	int size = packet.size + sizeof(RTP_FIXED_HEADER);
	m_buffer.resize(size);
	data = m_buffer.data();

	int ret = send(data, size, ms);
	if (ret > 0)
	{
		if (ret == size)
		{
			m_buffer.clear();
			done = true;
		}
		else
		{
			m_offset = ret;
		}
	}

	return done;
}

bool TcpsRtpSocket::recv(RtpPacketHandler handler, int ms)
{
	if (!checkConnect())
	{
		return false;
	}

	if (!m_socketPair.checkReadable(m_connSocket.getHandle(), ms))
	//if (!m_connSocket.checkReadable(ms))
	{
		return false;
	}

	uint8_t* data = m_recvBuffer.data();
	int size = m_recvBuffer.size();
	uint8_t* buf = data + size;

	int hdrSize = sizeof(int32_t) + sizeof(RTP_FIXED_HEADER);
	if (size < hdrSize)
	{
		int32_t toRead = hdrSize - size;
		int ret = m_connSocket.receive((char*)buf, toRead, 0);
		if (ret == 0)
		{
			setState(kDisconnected);
			m_recvBuffer.clear();
			return false;
		}
		else if (ret < 0)
		{
			return false;
		}
		else
		{
			m_recvBuffer.resize(size + ret);
			if (ret == toRead)
			{
				int32_t length = *((int32_t*)data);
				length = ntohl(length);

				m_recvBuffer.ensure(length);

				data = m_recvBuffer.data();
				size = m_recvBuffer.size();
				buf = data + size;
			}
			else
			{
				return false;
			}
		}
	}

	int32_t length = *((int32_t*)data);
	length = ntohl(length);

	int toRead = length - 4 - size;
	int ret = m_connSocket.receive((char*)buf, toRead, 0);
	if (ret == 0)
	{
		setState(kDisconnected);
		m_recvBuffer.clear();
		return false;
	}
	else if (ret < 0)
	{
		return false;
	}
	else
	{
		m_recvBuffer.resize(size + ret);

		if (ret == toRead)
		{
			data = m_recvBuffer.data() + 4;
			size = length;

			RTP_FIXED_HEADER* header = (RTP_FIXED_HEADER*)data;

			data += sizeof(RTP_FIXED_HEADER);
			size -= sizeof(RTP_FIXED_HEADER);

			RtpPacket pkt;
			memset(&pkt, 0, sizeof(pkt));
			pkt.pt = header->payload;
			pkt.mark = header->marker;
			pkt.ts = header->timestamp;
			pkt.data = data;
			pkt.size = size;

			handler(pkt);

			m_recvBuffer.clear();

			return true;
		}
		else
		{
			return false;
		}
	}
}

bool TcpsRtpSocket::isConnected()
{
	return m_connSocket.isOpen();
}

bool TcpsRtpSocket::flushWrite(int ms)
{
	if (m_offset < 0 || (m_offset >= (int)m_buffer.size()))
	{
		return true;
	}

	uint8_t* data = m_buffer.data() + m_offset;
	int size = m_buffer.size() - m_offset;

	int ret = send(data, size, ms);
	if (ret > 0)
	{
		m_offset += ret;
	}
	return (m_offset >= (int)m_buffer.size());
}

/// 发送并检测断开
int TcpsRtpSocket::send(uint8_t* data, int size, int ms)
{
	int ret = m_connSocket.sendBlock((char*)data, size);
	if (ret == -1)
	{
		int error = comn::Socket::getLastError();
		if (error == ECONNRESET)
		{
			setState(kDisconnected);
			fireConnectionState(kDisconnected);
		}
	}
	return ret;
}

void TcpsRtpSocket::setState(State state)
{
	comn::AutoCritSec lock(m_cs);
	m_state = state;
}

RtpSocket::State TcpsRtpSocket::getState()
{
	comn::AutoCritSec lock(m_cs);
	return m_state;
}

int TcpsRtpSocket::run()
{
	int millsec = 1000;

	while (!m_canExit)
	{
		if (!m_listenSocket.checkReadable(millsec))
		{
			continue;
		}

		comn::SockAddr addr;
		comn::Socket sock = m_listenSocket.accept(addr);
		if (!sock.isOpen())
		{
			continue;
		}

		CLog::info("TcpsRtpSocket incoming. %s:%d\n", addr.getIPAddr(), addr.getPort());

		sock.trySetSendBufferSize(SEND_BUFFER_SIZE, MIN_BUFFER_SIZE);
		sock.trySetRecvBufferSize(SEND_BUFFER_SIZE, MIN_BUFFER_SIZE);

		if (m_connSocket.isOpen())
		{
			fireConnectionState(kDisconnected);

			comn::AutoCritSec lock(m_cs);
			m_connSocket.close();
		}

		{
			comn::AutoCritSec lock(m_cs);
			m_connSocket = sock;
		}

		fireConnectionState(kConnected);
	}
	return 0;
}

void TcpsRtpSocket::doStop()
{
	comn::Socket sock;
	sock.open(SOCK_STREAM);
	sock.setNonblock(true);
	sock.connect(comn::SockAddr("127.0.0.1", m_addr.getPort()));
	sock.close();
}

bool TcpsRtpSocket::checkConnect()
{
	return m_connSocket.isOpen();
}

void TcpsRtpSocket::abort()
{
    comn::Socket sock;
    sock.open(SOCK_STREAM);
    sock.setNonblock(true);
    sock.connect(comn::SockAddr("127.0.0.1", m_addr.getPort()));
    sock.close();
}

bool TcpsRtpSocket::waitForConnection(int ms)
{
    if (!m_listenSocket.checkReadable(ms))
    {
        return false;
    }

    comn::SockAddr addr;
    comn::Socket sock = m_listenSocket.accept(addr);
    if (!sock.isOpen())
    {
        return false;
    }

    CLog::info("TcpsRtpSocket incoming. %s:%d\n", addr.getIPAddr(), addr.getPort());

    sock.trySetSendBufferSize(SEND_BUFFER_SIZE, MIN_BUFFER_SIZE);
    sock.trySetRecvBufferSize(SEND_BUFFER_SIZE, MIN_BUFFER_SIZE);

    if (m_connSocket.isOpen())
    {
        fireConnectionState(kDisconnected);

        comn::AutoCritSec lock(m_cs);
        m_connSocket.close();
    }

    {
        comn::AutoCritSec lock(m_cs);
        m_connSocket = sock;
    }

    fireConnectionState(kConnected);
    return true;
}



} /* namespace av */
